package com.o2o.cleaning.month.platform.ebusiness_plat.kuaishou.shuang11

import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object _04_ShopExtract_Execute {

  def main(args: Array[String]) {

    val spark = SparkSession.builder()
      .master("local[*]")
      .appName("extractStop_douyin")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .config("cluster.name", "O2OElastic")
      .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .getOrCreate()
    val sc: SparkContext = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("ERROR")
    /*============================================================================================*/
    //注：如果代码中有的路径吧中含有'/*'   则使用'/**  */'注释的时候会影响注释的范围：即路径中的'/*'会参与匹配

    val platformArr = Array("kuaishou")
    val year = "2021"
    val month = "10"
    for (platform <- platformArr) {
      val orc_path = s"s3a://dws-data/g_data/${year}/shuang11/2021_11_1_11/kuaishou/"
      val df_all = spark.read.orc(orc_path)
//obs://dws-data/g_data/2021/shuang11/shop/2021_11_1_11/平台
      val shopPath_json = s"s3a://o2o-sourcedata-${year}/obs_result_shop/${year}/${month}/${platform}"
      val shopPath_orc = s"s3a://dws-data/g_data/${year}/shuang11/shop/2021_11_1_11/${platform}"
      extractStop(spark, df_all, shopPath_json, shopPath_orc)

      spark.stop()
      sc.stop()

    }

    def extractStop(spark: SparkSession, frame: DataFrame, shopPath_json: String, shopPath_orc: String): DataFrame = {
      frame.registerTempTable("FEB_5")
      val shop = spark.sql(
        """
          |select
          |shopid,
          |cast(sum(sellcount) as Long) totalSellCount,
          |cast(sum(salesamount) as decimal(20,2)) totalSalesAmount
          |from
          |FEB_5
          |group by
          |shopid
      """.
          stripMargin)
      shop.registerTempTable("shop")

      //a.shopType    0411去掉了
      val
      mttg_shop_all = spark.sql(
        """
          |select
          | a.address AS address
          |,a.administrative_region AS administrative_region
          |,a.aedzId AS aedzId
          |,a.city AS city
          |,a.city_grade AS city_grade
          |,a.city_origin AS city_origin
          |,a.name AS company_name
          |,a.district AS district
          |,a.district_origin AS district_origin
          |,a.economic_division AS economic_division
          |,a.goodratepercentage AS goodratepercentage
          |,a.if_city AS if_city
          |,a.if_district AS if_district
          |,a.latitude AS latitude
          |,a.longitude AS longitude
          |,a.name AS name
          |,a.province AS province
          |,a.regional_ID AS regional_ID
          |,a.registration_institution AS registration_institution
          |,a.rural_demonstration_counties AS rural_demonstration_counties
          |,a.rural_ecommerce AS rural_ecommerce
          |,a.the_belt_and_road_city AS the_belt_and_road_city
          |,a.the_belt_and_road_province AS the_belt_and_road_province
          |,a.the_yangtze_river_economic_zone_city AS the_yangtze_river_economic_zone_city
          |,a.the_yangtze_river_economic_zone_province AS the_yangtze_river_economic_zone_province
          |,a.town AS town
          |,a.urban_agglomerations AS urban_agglomerations
          |,a.allitemcount AS allitemcount
          |--,a.anchor_count AS anchor_count
          |,a.shopType AS BC_type
          |,a.platformId AS platformId
          |,a.platformName AS platformName
          |,a.shopId AS shopId
          |--,a.shopImages AS shopImages
          |,a.shopName AS shopName
          |,a.shopType AS shopType
          |,a.shopUrl AS shopUrl
          |,a.is_flagship AS is_flagship
          |-- ,a.is_brandShip AS is_brandShip
          |,a.timeStamp AS timeStamp
          |
          |,b.totalSellCount
          |,b.totalSalesAmount
          |
          |from FEB_5 a
          |left join
          |shop b
          |on a.shopid = b.shopid
      """.stripMargin)
        .dropDuplicates("shopid")

//      mttg_shop_all.repartition(4).write.mode(SaveMode.Overwrite).json(shopPath_json)
      mttg_shop_all.repartition(4).write.mode(SaveMode.Overwrite).orc(shopPath_orc)
      mttg_shop_all

    }
  }
}
